package com.myCompany.edu.realtime.app.dwd.db;

import com.myCompany.edu.realtime.app.BaseSQLApp;
import com.myCompany.edu.realtime.common.Constant;
import com.myCompany.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd08_DwdTradePaySuc extends BaseSQLApp{
    public static void main(String[] args) {
        new Dwd08_DwdTradePaySuc().init(
                3008,
                2,
                "Dwd08_DwdTradePaySuc"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        //读取 ods_db
        readOdsDb(tEnv,"Dwd08_DwdTradePaySuc");

        Table paymentInfo = tEnv.sqlQuery("select " +
                "`data`['order_id'] order_id, " +
                "ts, " +
                "`pt` " +
                "from ods_db " +
                "where `database` = 'edu' " +
                "and `table` = 'payment_info'");

        tEnv.createTemporaryView("payment_info",paymentInfo);
       // paymentInfo.execute().print();


        tEnv.executeSql("create table dwd_trade_order_detail( " +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3))"+ SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,"Dwd08_DwdTradePaySuc"));
        //两张表join
        Table result = tEnv.sqlQuery("select " +
                "user_id, " +
                "pi.ts, " +
                "od.row_op_ts row_op_ts " +
                "from payment_info pi " +
                "join dwd_trade_order_detail od " +
                "on pi.order_id = od.order_id");
        //创建表与 topic 关联
        tEnv.createTemporaryView("result",result);
//        result.execute().print();
        tEnv.executeSql("create table dwd_trade_pay_suc( " +
                "user_id string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_PAY_SUC));
        //写出到输出表
        result.executeInsert("dwd_trade_pay_suc");

    }
}
